package com.sam.kafka.serialization;

import com.sam.kafka.entity.User;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author: wm.xue
 * @date: 2021/11/5 18:32
 * @description:
 */
public class UserSerializer implements Serializer<User> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // nothing to do
    }

    @Override
    public byte[] serialize(String topic, User data) {

        if(data == null) return null;

        Integer userId = data.getUserId();
        String userName = data.getUserName();

        byte[] userNameBytes = userName.getBytes(StandardCharsets.UTF_8);
        int length = userNameBytes.length;

        ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);
        buffer.putInt(userId);
        buffer.putInt(length);
        buffer.put(userNameBytes);

        return buffer.array();
    }

    @Override
    public void close() {
        // nothing to do
    }
}
